package com.steve.bigdata.flinkmonitor.assist;

import com.steve.bigdata.flinkmonitor.model.Event;
import org.apache.flink.api.common.functions.MapFunction;

/**
 * @Author: STEVE
 * @Description: 辅助类 - Kafka原始数据解析器
 * 功能：将Kafka中的字符串消息转换为Event对象
 * 预期输入格式："userId,timestamp,value"（如："user123,1625097600000,150.0"）
 * @since: 2025-07-10
 */
public class EventParser implements MapFunction<String, Event> {

    /**
     * 解析字符串为Event对象
     *
     * @param value Kafka消息字符串
     * @return 解析后的Event对象
     * @throws Exception 当数据格式不符合预期时抛出异常
     */
    @Override
    public Event map(String value) throws Exception {
        try {
            String[] parts = value.split(",");  // 按逗号分割字段
            if (parts.length != 3) {
                throw new IllegalArgumentException("Invalid message format. Expected 3 fields");
            }
            // 构建Event对象
            return new Event(
                    parts[0].trim(),    // userId
                    Long.parseLong(parts[1].trim()),    // timestamp (milliseconds)
                    Double.parseDouble(parts[2].trim()) // value
            );
        } catch (Exception e) {
            // 生产环境应添加监控指标（如解析失败计数器）
            throw new RuntimeException("Failed to parse event: " + value, e);
        }
    }

}
